package cgl.narada.transport;

import java.util.Enumeration;
import java.util.Hashtable;

public class BrokerConnections {

    private Hashtable singleConnectionsToUnit; // <naradaBrokeringId, linkId>

    private Hashtable multipleConnectionsToUnit;// <naradaBrokeringId,

    // (connectionId, LinkId)

    private Hashtable linkAndBrokerIDs; // <linkId, brokerId>

    private Hashtable linkAndConnectionIDs; // < linkId, connectionId>

    private Hashtable connectionAndBrokerIDs; // <connectionId, brokerId>

    private String moduleName = "BrokerConnections: ";

    private TransportHandler transportHandler;

    private boolean debug = false;


    public BrokerConnections(TransportHandler transportHandler) {
        this.transportHandler = transportHandler;
        singleConnectionsToUnit = new Hashtable();
        multipleConnectionsToUnit = new Hashtable();
        linkAndBrokerIDs = new Hashtable();
        linkAndConnectionIDs = new Hashtable();
        connectionAndBrokerIDs = new Hashtable();
    }


    public synchronized void mapLinkToBrokerNode(
                                                 String linkId,
                                                 Object naradaBrokeringId,
                                                 Object brokerConnectionIdentifier)
        throws TransportException {

        String errorReport = null;
        if (linkAndConnectionIDs.containsKey(linkId)) {
            errorReport =
                moduleName + "The link [" + linkId
                    + "] has been mapped previously to connecction-ID ["
                    + brokerConnectionIdentifier + "]. No further processing!";
            throwTransportException(errorReport);
        }

        boolean multipleConnectionsUpdated =
            checkAndUpdateMultipleConnectionsToUnit(linkId, naradaBrokeringId,
                                                    brokerConnectionIdentifier);
        if (!multipleConnectionsUpdated) {
            checkAndUpdateSingleConnectionsToUnit(linkId, naradaBrokeringId,
                                                  brokerConnectionIdentifier);
        }
    }


    private void checkAndUpdateSingleConnectionsToUnit(
                                                       String linkId,
                                                       Object naradaBrokeringId,
                                                       Object brokerConnectionIdentifier)
        throws TransportException {

        if (!singleConnectionsToUnit.containsKey(naradaBrokeringId)) {
            singleConnectionsToUnit.put(naradaBrokeringId, linkId);
            updateTables(linkId, naradaBrokeringId, brokerConnectionIdentifier);
            String report =
                moduleName + "Mapping of Broker [" + naradaBrokeringId
                    + "] and link [" + linkId + "] complete!";
            System.out.println(report);
            return;
        }

        /** This implies that this is the second connection to the unit. */

        /**
         * We first need to remove existing unit connection, and move into
         * multiple connections list before putting the one in the arguments of
         * this method.
         */
        Object firstLinkId = singleConnectionsToUnit.get(naradaBrokeringId);
        Object firstConnectionId = linkAndConnectionIDs.get(firstLinkId);

        singleConnectionsToUnit.remove(naradaBrokeringId);

        Hashtable connectionList = new Hashtable();
        connectionList.put(firstConnectionId, firstLinkId);

        multipleConnectionsToUnit.put(naradaBrokeringId, connectionList);

        /**
         * Now put the broker connection listed in the arguments of this method
         * into multipleConnections Table.
         */
        checkAndUpdateMultipleConnectionsToUnit(linkId, naradaBrokeringId,
                                                brokerConnectionIdentifier);
    }


    /**
     * Check to see if there are already multiple connections to this unit.
     * 
     * @param linkId
     * @param naradaBrokeringId
     * @param brokerConnectionIdentifier
     * @return
     */
    private boolean checkAndUpdateMultipleConnectionsToUnit(
                                                            String linkId,
                                                            Object naradaBrokeringId,
                                                            Object brokerConnectionIdentifier)
        throws TransportException {

        boolean updated = false;
        if (multipleConnectionsToUnit.containsKey(naradaBrokeringId)) {
            Hashtable multipleConnections =
                (Hashtable) multipleConnectionsToUnit.get(naradaBrokeringId);
            if (multipleConnections.containsKey(brokerConnectionIdentifier)) {
                String errorReport =
                    moduleName + "This should not happen. "
                        + "The connection identifier "
                        + brokerConnectionIdentifier
                        + " of the new unit is previously known to this NODE";
                throwTransportException(errorReport);
            }
            updated = true;
            multipleConnections.put(brokerConnectionIdentifier, linkId);

            String report =
                moduleName + "There are multiple connections to the unit ["
                    + naradaBrokeringId + "]. These include the following "
                    + "connection/linkID pairs " + multipleConnections;

            System.out.println(report);
            updateTables(linkId, naradaBrokeringId, brokerConnectionIdentifier);
        }

        return updated;
    }


    private void updateTables(String linkId, Object naradaBrokeringId,
                              Object brokerConnectionIdentifier) {
        linkAndBrokerIDs.put(linkId, naradaBrokeringId);
        linkAndConnectionIDs.put(linkId, brokerConnectionIdentifier);
        connectionAndBrokerIDs.put(brokerConnectionIdentifier,
                                   naradaBrokeringId);
    }


    public void sendData(byte[] data, Object naradaBrokeringId)
        throws TransportException {
        String report = "";
        if (singleConnectionsToUnit.containsKey(naradaBrokeringId)) {
            String linkId =
                (String) singleConnectionsToUnit.get(naradaBrokeringId);
            if (debug) {
                report = moduleName + "Will send data over link " + linkId;
                System.out.println(report);
            }
            transportHandler.sendData(data, linkId);
            return;
        }

        if (multipleConnectionsToUnit.containsKey(naradaBrokeringId)) {
            Hashtable connections =
                (Hashtable) multipleConnectionsToUnit.get(naradaBrokeringId);
            Enumeration connectionIDKeys = connections.keys();

            report =
                moduleName + "Will send data over the following ("
                    + +connections.size() + ")links \n";
            int _linkNum = 0;
            while (connectionIDKeys.hasMoreElements()) {
                Object _connectionIdentifier = connectionIDKeys.nextElement();
                String _linkId =
                    (String) connections.get(_connectionIdentifier);
                _linkNum++;
                transportHandler.sendData(data, _linkId);
                if (debug) {
                    report += ("(" + _linkNum + ")" + _linkId + "\n");
                }
            }
            if (debug) {
                System.out.println(report);
            }
            return;
        }

        report =
            "There are no links to the specified broker =>" + naradaBrokeringId;
        throwTransportException(report);
    }


    /**
     * Process the loss of connection to a linkId. Return
     * 
     * @param linkId
     * @return Object the connection identifier associated with the link in 
     * question.
     */
    public synchronized Object processLossOfConnection(String linkId)
        throws TransportException {
        String errorReport = null;
        if (!linkAndBrokerIDs.containsKey(linkId)) {
            errorReport =
                moduleName + "The BROKEN Link with ID (" + linkId
                    + ") does NOT have an entry in the linkAndBrokerIDs table";
            throwTransportException(errorReport);
        }

        Object brokerId = linkAndBrokerIDs.remove(linkId);
        Object connectionId = linkAndConnectionIDs.remove(linkId);

        if (connectionId != null) {
            connectionAndBrokerIDs.remove(connectionId);
        } else {
            errorReport =
                moduleName + "ConnectionID  associated with link (" + linkId
                    + ") is not present in the [linkAndConnectionIDs] table";
            throwTransportException(errorReport);
        }

        if (singleConnectionsToUnit.containsKey(brokerId)) {
            singleConnectionsToUnit.remove(brokerId);
            System.out.println(moduleName + "Removed broker [" + brokerId
                + "] from (singleConnectionsToUnit) Table");
        }

        if (multipleConnectionsToUnit.containsKey(brokerId)) {
            Hashtable connections =
                (Hashtable) multipleConnectionsToUnit.get(brokerId);

            if (connections.containsKey(connectionId)) {
                connections.remove(connectionId);
            }

            if (connections.size() == 1) {
                Hashtable connectionInfo =
                    (Hashtable) multipleConnectionsToUnit.remove(brokerId);
                Enumeration links = connectionInfo.elements();
                Object linkIdToRelocate = links.nextElement();
                singleConnectionsToUnit.put(brokerId, linkIdToRelocate);

                String report =
                    moduleName + "There is NOW only ONE connection" + " to ("
                        + brokerId + ") . This has now been moved to the "
                        + "singleConnectionsToUnit table";
                System.out.println(report);
            }

        }

        return connectionId;
    }


    private void throwTransportException(String errorReport)
        throws TransportException {
        TransportException transportException =
            new TransportException(errorReport);
        throw transportException;
    }


    public static void main(String[] args) {
        TransportHandler transportHandler = null;
        BrokerConnections multipleConnections =
            new BrokerConnections(transportHandler);

        String linkId1 = "tcp://mercury.ucs.indiana.edu:3045";
        String linkId2 = "tcp://venus.ucs.indiana.edu:3045";
        String linkId3 = "tcp://earth.ucs.indiana.edu:3045";
        String linkId4 = "tcp://mars.ucs.indiana.edu:3045";
        String linkId5 = "tcp://jupiter.ucs.indiana.edu:3045";
        String linkId6 = "tcp://saturn.ucs.indiana.edu:3045";

        String nbId1 = "NB-One";
        String nbId2 = "NB-Two";

        String connectionId1 = "C1";
        String connectionId2 = "C2";
        String connectionId3 = "C3";
        String connectionId4 = "C4";
        String connectionId5 = "C5";
        String connectionId6 = "C6";
        String connectionId7 = "C7";

        try {

            multipleConnections.mapLinkToBrokerNode(linkId1, nbId1,
                                                    connectionId1);
            multipleConnections.mapLinkToBrokerNode(linkId2, nbId1,
                                                    connectionId2);
            multipleConnections.mapLinkToBrokerNode(linkId3, nbId1,
                                                    connectionId3);
            multipleConnections.mapLinkToBrokerNode(linkId4, nbId2,
                                                    connectionId4);
            multipleConnections.mapLinkToBrokerNode(linkId5, nbId2,
                                                    connectionId5);
            multipleConnections.mapLinkToBrokerNode(linkId6, nbId1,
                                                    connectionId6);

            multipleConnections.processLossOfConnection(linkId4);
            multipleConnections.processLossOfConnection(linkId3);

            byte[] data = "Test Data".getBytes();
            multipleConnections.sendData(data, nbId1);
            multipleConnections.sendData(data, nbId2);
        } catch (TransportException transEx) {
            transEx.printStackTrace();
        }
    }
}
